package cn.touna.jss.modules.credit.consumer;

import cn.touna.jss.common.cache.HbaseComponent;
import cn.touna.jss.common.entity.CreditQueryRequest;
import cn.touna.jss.modules.credit.service.AuthHistoryService;
import cn.touna.jss.modules.credit.service.CreditQueryService;
import cn.touna.risk.api.domain.Response;
import cn.touna.risk.api.utils.JSONUtil;
import cn.touna.risk.api.utils.ServiceContext;
import com.alibaba.fastjson.JSONException;
import com.alibaba.fastjson.JSONObject;
import com.touna.loan.log.service.TracerContextUtil;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

/**
 * @Author:wuwc
 * @Description:
 * @Date:2018/4/17 13:04
 */
@Component
public class CreditHistoryConsumer {
    private static final Logger logger = LogManager.getLogger(CreditHistoryConsumer.class);

    @Autowired
    protected HbaseComponent hbaseComponent;

    @Autowired
    private CreditQueryService creditQueryService;

    @Autowired
    private AuthHistoryService authHistoryService;

    //@KafkaListener(topics = {"${kafka.topic.credithistory}"})
    public void listern(ConsumerRecord<?, ?> record, Acknowledgment ack) {
        String traceId = record.key() == null ? ServiceContext.genUniqueId() : record.key().toString();
        TracerContextUtil.attachTraceId(traceId);
        logger.info("REQ【调用记录消费】,traceId:{},topic:{},partition:{}", record.key(), record.topic(), record.partition());
        if (null != record.value()) {
            boolean isSuccess = true;
            try {
                JSONObject msgJson = JSONObject.parseObject(record.value().toString());
                if (msgJson == null) {
                    return;
                }
                JSONObject requestJson = msgJson.getJSONObject("request");
                JSONObject respJson = msgJson.getJSONObject("response");
                CreditQueryRequest queryRequest = JSONUtil.jsonToObject(requestJson, CreditQueryRequest.class);
                Response response = JSONUtil.jsonToObject(respJson, Response.class);
                saveQueryHistory(queryRequest, response);
            } catch (JSONException e) {
                logger.error("REQ【调用记录消费JSON解析异常】,msg: {}", e);
            } catch (Exception e) {
                isSuccess = false;
                logger.error("REQ【调用记录消费异常】,msg: {}", e);
            } finally {
                if (isSuccess) {
                    ack.acknowledge();
                }
            }
        }
    }

    /**
     * 保存查询历史
     *
     * @param request
     */
    private void saveQueryHistory(CreditQueryRequest request, Response response) {
        try {
            //TODO:资信报告上线后去掉mysql的存储
            creditQueryService.saveQueryHistory(request, response);
            //保存认证历史
            authHistoryService.saveAuthHistory(request, response);
        } catch (Exception e) {
            //LogUtil.logError("保存查询记录异常", request, true, e);
        }
    }
}
